package com.my.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;


import java.time.Duration;
import java.util.Arrays;
import java.util.Iterator;
import java.util.Properties;

/**
 * Kafka消费者
 */
public class KafkaConsumerExample implements Runnable {

    private final KafkaConsumer<String, Object> kafkaConsumer;

    private final String topic;


    /**
     * enable.auto.commit 默认为true
     * auto.commit.interval.ms 默认为5000 ms (5s)
     * max.poll.records 默认为500
     * fetch.max.bytes 默认为52428800 bytes (50Mib).
     * @param topic
     */
    public KafkaConsumerExample(String topic) {

        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094");
        properties.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "GROUP_ID0");
        //earliest, latest, none, anything else
        properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        //一次poll records最大数量
        //properties.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "15");
        //一次fetch最大接收数据量
        //properties.setProperty(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "");
        this.kafkaConsumer = new KafkaConsumer<>(properties);
        kafkaConsumer.seekToBeginning(kafkaConsumer.assignment());
        this.topic = topic;
        this.kafkaConsumer.subscribe(Arrays.asList(topic));

    }

    @Override
    public void run() {

        Duration duration = Duration.ZERO;
        while (true) {

            ConsumerRecords<String, Object> records = kafkaConsumer.poll(duration);
            if(records.isEmpty()){
                try{
                    Thread.sleep(1000);
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
            Iterator<ConsumerRecord<String, Object>> recordIterator = records.iterator();
            while (recordIterator.hasNext()){
                ConsumerRecord<String, Object> record = recordIterator.next();
                System.out.println("From topic: " + record.topic() + "  key is: " + record.key() +"  value is: " + record.value().toString());
            }
        }
    }

    public static void main(String[] args) {

            KafkaConsumerExample consumer = new KafkaConsumerExample("" +
                    "");
            Thread thread = new Thread(consumer);
            thread.start();
        System.out.println("hahhaha");
    }
}
